package cn.itcast.tags.models.ml

import cn.itcast.tags.config.ModelConfig
import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.{MLModelTools, TagTools}
import org.apache.spark.ml.clustering.KMeansModel
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.ml.linalg
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 开发标签模型（挖掘类型标签）：用户活跃度RFE模型
 * Recency: 最近一次访问时间,用户最后一次访问距今时间
 * Frequency: 访问频率,用户一段时间内访问的页面总次数,
 * Engagements: 页面互动度,用户一段时间内访问的独立页面数,也可以定义为页面 浏览量、下载量、 视频播放数量等
 */
class RfeModel extends AbstractModel("用户活跃度模型RFE模型", ModelType.ML) {
  /*
    367 用户活跃度
      368	非常活跃		0
      369	活跃		    1
      370	不活跃	   	2
      371	非常不活跃	3
   */
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val spark: SparkSession = businessDF.sparkSession
    import org.apache.spark.sql.functions._
    import spark.implicits._

    // TODO: 1. 从业务数据中计算R、F、E值
    val rfeDF: DataFrame = businessDF
      .groupBy($"global_user_id") // 按照用户分组
      // 使用函数计算R、F、E值
      .agg(
        max($"log_time").as("last_time"),
        count($"loc_url").as("frequency"),
        countDistinct($"loc_url").as("engagements")
      ).
      select(
        $"global_user_id".as("userId"),
        // 计算R值
        datediff(
          date_sub(current_timestamp(), 600), $"last_time"
        ).as("recency"),
        $"frequency", $"engagements"
      )
    /*
      root
       |-- userId: string (nullable = true)
       |-- recency: integer (nullable = true)
       |-- frequency: long (nullable = false)
       |-- engagements: long (nullable = false)
     */
    // rfeDF.printSchema()
    // rfeDF.show(50, truncate = false)

    // TODO: 2. 按照规则，给RFE值打分: Score
    /*
      R:0-15天=5分，16-30天=4分，31-45天=3分，46-60天=2分，大于61天=1分
      F:≥400=5分，300-399=4分，200-299=3分，100-199=2分，≤99=1分
      E:≥250=5分，200-249=4分，150-199=3分，149-50=2分，≤49=1分
    */
    // R 打分条件表达式
    val rWhen = when(col("recency").between(1, 15), 5.0)
      .when(col("recency").between(16, 30), 4.0)
      .when(col("recency").between(31, 45), 3.0)
      .when(col("recency").between(46, 60), 2.0)
      .when(col("recency").geq(61), 1.0) //
    // F 打分条件表达式
    val fWhen = when(col("frequency").leq(99), 1.0)
      .when(col("frequency").between(100, 199), 2.0)
      .when(col("frequency").between(200, 299), 3.0)
      .when(col("frequency").between(300, 399), 4.0)
      .when(col("frequency").geq(400), 5.0)
    // E 打分条件表达式
    val eWhen = when(col("engagements").lt(49), 1.0)
      .when(col("engagements").between(50, 149), 2.0)
      .when(col("engagements").between(150, 199), 3.0)
      .when(col("engagements").between(200, 249), 4.0)
      .when(col("engagements").geq(250), 5.0)
    val rfeScoreDF: DataFrame = rfeDF.select(
      $"userId",
      rWhen.as("r_score"),
      fWhen.as("f_score"),
      eWhen.as("e_score")
    )
    // rfeScoreDF.printSchema()
    // rfeScoreDF.show()

    // TODO: 3. 计算rfe的值，使用KMeans聚类算法训练，得到最佳模型，并预测
    // 3.1 组合R\F\E列为特征值features
    val assembler: VectorAssembler = new VectorAssembler()
      .setInputCols(Array("r_score", "f_score", "e_score"))
      .setOutputCol("features")
    val rfeFeaturesDF: DataFrame = assembler.transform(rfeScoreDF)

    // 3.2 获取KMeans模型
    val kMeansModel: KMeansModel = MLModelTools
      .loadModel(rfeFeaturesDF, "rfe", this.getClass).asInstanceOf[KMeansModel]

    // 3.3 模型预测
    val predictionDF: DataFrame = kMeansModel.transform(rfeFeaturesDF)
    // predictionDF.printSchema()
    // predictionDF.show(50, truncate = false)

    // TODO: 4. 打标签
    // 4.1 获取聚类模型中簇中心及索引
    val clusterCenters: Array[linalg.Vector] = kMeansModel.clusterCenters
    // 4.2 类簇中心点对应标签ID
    val indexTagMap = TagTools.convertIndexMap(clusterCenters, tagDF)
    val indexTagMapBroadcast = spark.sparkContext.broadcast(indexTagMap)
    // 4.3. 关联聚类预测值，将prediction转换为tagName
    val index_to_tag = udf(
      (prediction: Int) => indexTagMapBroadcast.value(prediction)
    )
    val modelDF: DataFrame = predictionDF
      .select(
        $"userId",
        index_to_tag($"prediction").as("rfe")
      )
    modelDF.show(100, truncate = false)

    // 7. 返回标签数据
    modelDF
  }

}

object RfeModel {
  def main(args: Array[String]): Unit = {
    val tagModel = new RfeModel()
    tagModel.executeModel(367L)
  }
}
